package com.boot.cluster.utils;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;

public class KryoSerializer<T> {
	public static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
	private static final ThreadLocal<Kryo> local = ThreadLocal.withInitial(Kryo::new);
	private static Logger logger = LoggerFactory.getLogger(KryoSerializer.class);
	private Class<T> clazz;


	public KryoSerializer(Class<T> clazz) {
		this.clazz = clazz;
	}

	public byte[] serialize(T t) {
		if (t == null) {
			return EMPTY_BYTE_ARRAY;
		}

		Kryo kryo = local.get();
		kryo.setReferences(false);
		kryo.register(this.clazz);

		try (ByteArrayOutputStream os = new ByteArrayOutputStream();
			 Output output = new Output(os)) {
			kryo.writeClassAndObject(output, t);
			output.flush();
			return os.toByteArray();
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}

		return EMPTY_BYTE_ARRAY;
	}

	public T deserialize(byte[] bytes) {
		if ((bytes == null) || (bytes.length <= 0)) {
			return null;
		}
		Kryo kryo = local.get();
		kryo.setReferences(false);
		kryo.register(this.clazz);
		try (Input input = new Input(bytes)) {
			return (T) kryo.readClassAndObject(input);
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
		return null;
	}


}
